package com.xdja.kafka.hdfs.sink;
import java.io.IOException;
import java.util.Random;

import com.alibaba.fastjson.JSON;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.GroupFactory;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.ParquetReader.Builder;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.example.GroupWriteSupport;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;

public class ParquetHandle {
    /**
     * 读取对象属性
     * @param inPath
     * @throws Exception
     */
    static void parquetReadProperty(String inPath) throws Exception{
        GroupReadSupport readSupport = new GroupReadSupport();
        Builder<Group> reader= ParquetReader.builder(readSupport, new Path(inPath));
        ParquetReader<Group> build=reader.build();
        Group line = null;
        while((line = build.read())!=null) {
            Group time= line.getGroup("time", 0);
            String city = line.getString("city", 0);
            String ip = line.getString("ip", 0);
            Integer ttl = time.getInteger("ttl", 0);
            String ttl2 = time.getString("ttl2", 0);

            System.out.println("city:" + city + ", ip: " + ip + ", ttl:" + ttl + ", ttl2:" + ttl2);

        }

    }

    /**
     * 读取一行
     * @param inPath
     * @throws Exception
     */
    public static void parquetReadLine(String inPath) throws Exception{
        GroupReadSupport readSupport = new GroupReadSupport();
        ParquetReader<Group> reader = new ParquetReader<Group>(new Path(inPath),readSupport);
        Group line=null;
        while((line = reader.read()) != null){
            System.out.println(JSON.toJSONString(line));
        }
        System.out.println("读取结束");

    }
    /**
     *
     * @param outPath　　输出Parquet格式
     * @throws IOException
     */
    static void parquetWrite(String outPath) throws IOException{
        MessageType schema = MessageTypeParser.parseMessageType(
      "message Pair {\n" +
                 " required binary city (UTF8);\n" +
                 " required binary ip (UTF8);\n" +
                 " repeated group time {\n" +
                       " required int32 ttl;\n"+
                       " required binary ttl2;\n"+
                 "}\n"+
            "}");
        GroupFactory factory = new SimpleGroupFactory(schema);
        Path path = new Path(outPath);
        Configuration configuration = new Configuration();
        GroupWriteSupport writeSupport = new GroupWriteSupport();
        writeSupport.setSchema(schema,configuration);
        ParquetWriter<Group> writer = new ParquetWriter<Group>(path,configuration,writeSupport);
        Random r=new Random();
        String[] strs = new String[]{"西安", "192.168.35.204"};
        Group group = factory.newGroup()
                .append("city",strs[0])
                .append("ip",strs[1]);
        Group tmpG =group.addGroup("time");
        tmpG.append("ttl", r.nextInt(9) + 1);
        tmpG.append("ttl2", r.nextInt(9) + "_a");
        writer.write(group);

        strs = new String[]{"郑州", "192.168.35.205"};
        group = factory.newGroup()
                .append("city",strs[0])
                .append("ip",strs[1]);
        tmpG =group.addGroup("time");
        tmpG.append("ttl", r.nextInt(9) + 1);
        tmpG.append("ttl2", r.nextInt(9) + "_a");
        writer.write(group);
        System.out.println("write end");

        writer.close();
    }

    public static void main(String[] args) throws Exception {
        String path = "hdfs://192.168.35.203:8020/N1ydjw/wjy/23.txt";
        //写入
//        parquetWrite(path);
        //读取对象属性
//        parquetReadProperty(path);
        //读取一行
        parquetReadLine(path);
    }
}